package com.microstrategy.database.hbase.datasource;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.HServerInfo;
import org.apache.hadoop.hbase.HServerLoad;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.HServerLoad.RegionLoad;
import org.apache.hadoop.hbase.client.HBaseAdmin;

import util.hbaseutils;

import com.microstrategy.database.hbase.model.RegionOfTableOnEachRSServer;

public class AdminDataSource {
	private Configuration conf;
	private HBaseAdmin admin;

	public AdminDataSource(Configuration conf) {
		this.conf = conf;
		try {
			admin = new HBaseAdmin(this.conf);
		} catch (MasterNotRunningException e) {
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			e.printStackTrace();
		}
	}

	public List<RegionOfTableOnEachRSServer> getRTRS() {
		List<RegionOfTableOnEachRSServer> rlist = new ArrayList<RegionOfTableOnEachRSServer>();
		try {
			// Cluster status
			ClusterStatus status = admin.getClusterStatus();
			// each Hserver information
			Collection<ServerName> serverInfos = status.getServers();
			for (ServerName si : serverInfos) {
				HServerLoad sload = status.getLoad(si);

				// each Region information of that server
				Map<byte[], RegionLoad> regionLoads = sload.getRegionsLoad();
				for (byte[] regionnm : regionLoads.keySet()) {
					RegionOfTableOnEachRSServer lRtrs = new RegionOfTableOnEachRSServer();
					// time stamp
					Timestamp ts = new Timestamp(new Date().getTime());
					lRtrs.setTimestamp(ts.toString());
					// set region server name
					lRtrs.setRsname(si.getServerName());
					RegionLoad rl = regionLoads.get(regionnm);
					String rname = rl.getNameAsString();
					// the tablename
					lRtrs.setTablename(hbaseutils
							.getTableNameInRegionName(rname));
					// the region name
					lRtrs.setRegionname(rname);
					// the memstore size
					lRtrs.setMemstoresize(rl.getMemStoreSizeMB() + "");
					// total store size
					lRtrs.setStoresize(rl.getStorefileSizeMB() + "");
					// # of store numbers
					lRtrs.setStorenumber(rl.getStorefiles() + "");
					// store file index size
					lRtrs.setStorefileindexsize(rl.getStorefileIndexSizeMB()
							+ "");
					rlist.add(lRtrs);
				}
			}
			return rlist;

		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}
}
